package analyse;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.*;
import java.util.Map.Entry;

/**
 * 使用消费者实现的解析器
 * 采样间隔通过读取value中的采样发生时间控制
 *
 * @author pf
 */
@Deprecated
public class SamplerAnalyse implements Runnable {

	private KafkaConsumer<String, String> consumer;
	private String brokerUrl;
	private String topic;
	private int interval;
	public SamplerAnalyse(String brokerUrl, String topic, int interval) {

		this.brokerUrl = brokerUrl;
		this.topic = topic;
		this.interval = interval;

		creatConsumer();
	}

	public static void main(String[] args) {

		new Thread(new SamplerAnalyse("172.18.84.135:9092", "test", 10)).start();
	}

	public void creatConsumer() {

		Properties props = new Properties();
		// 集群地址，多个地址用"，"分隔
		props.put("bootstrap.servers", brokerUrl);
		// 设置消费者的group id
		props.put("group.id", UUID.randomUUID().toString());
		// 如果为真，consumer所消费消息的offset将会自动的同步到zookeeper。如果消费者死掉时，由新的consumer使用继续接替
		props.put("enable.auto.commit", "true");
		// consumer向zookeeper提交offset的频率
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		// 反序列化
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

		// 创建消费者
		this.consumer = new KafkaConsumer<String, String>(props);

		// 订阅topic，可以为多个用,隔开，此处订阅了"test-partition-1", "test"这两个主题
		consumer.subscribe(Arrays.asList(topic));

	}

	// 持续监听：
	@Override
	public void run() {

		System.out.println("开始监听");
		HashMap<String, ArrayList<JSONObject>> samplers = new HashMap<String, ArrayList<JSONObject>>();

		long startStamp = 0L;
		long endStamp = 0L;
		boolean flag = true;

		while (true) {
			// poll频率，超时时间1000ms
			ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1000));
			for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {

				String key = consumerRecord.key();
				String value = consumerRecord.value();
				JSONObject valueJson = JSONObject.parseObject(value);

				ArrayList<JSONObject> tempList = samplers.get(key);
				if (tempList == null) {
					tempList = new ArrayList<JSONObject>();
				}

				tempList.add(valueJson);
				samplers.put(key, tempList);

				// 记录开始时间
				if (flag) {
					startStamp = valueJson.getLongValue("endTimeStamp");
					flag = false;
				}

				// 更新当前时间
				endStamp = valueJson.getLongValue("endTimeStamp");

				// 当前时间到开始时间大于预定间隔，开始解析收集到的内容
				if ((endStamp - startStamp) >= interval) {

					// 调用解析方法，此处使用多线程
					long starttt = System.currentTimeMillis();
					String tempsss = analyse(samplers, endStamp).toJSONString();
					long enddddd = System.currentTimeMillis();
					System.out.println("耗时：" + (enddddd - starttt) + " 结果：" + tempsss);

					// 清理数据
					samplers.clear();

					// 重新记录开始时间
					flag = true;
				}

			}
		}
	}

	public JSONArray analyse(HashMap<String, ArrayList<JSONObject>> samplers, long endStamp) {

		JSONArray resArry = new JSONArray();

		JSONObject total = new JSONObject();

		double TOTAL_sendBytes = 0.0;
		double TOTAL_receiveBytes = 0.0;
		int TOTAL_threadCount = 0;
		int TOTAL_successCount = 0;
		int TOTAL_failCount = 0;
		int TOTAL_totalDelay = 0;
		int TOTAL_avgDelay = 0;
		int TOTAL_qps = 0;
		int TOTAL_length = 0;
		int TOTAL_length_level1 = 0;
		int TOTAL_length_level2 = 0;
		int TOTAL_length_level3 = 0;
		ArrayList<Integer> TOTAL_delays = new ArrayList<Integer>();

		for (Entry<String, ArrayList<JSONObject>> entry : samplers.entrySet()) {

			String key = entry.getKey();
			ArrayList<JSONObject> value = entry.getValue();

			JSONObject resObj = new JSONObject();

			double sendBytes = 0.0;
			double receiveBytes = 0.0;
			int threadCount = 0;
			int successCount = 0;
			int failCount = 0;
			int totalDelay = 0;
			int avgDelay = 0;
			int qps = 0;
			int length = value.size();
			int length_level1 = length * 90 / 100 - 1;
			int length_level2 = length * 95 / 100 - 1;
			int length_level3 = length * 99 / 100 - 1;

			if (length_level1 <= 0) {
				continue;
			}

			ArrayList<Integer> delays = new ArrayList<Integer>();

			for (JSONObject jsonObject : value) {

				// 计算这段时间内的采样数
				if (jsonObject.getBooleanValue("success")) {
					successCount = successCount + jsonObject.getIntValue("sampleCount");
				} else {
					failCount = failCount + jsonObject.getIntValue("sampleCount");
				}

				// 计算这段时间内的总耗时
				int delay = jsonObject.getIntValue("delayMS");
				totalDelay = totalDelay + delay;
				delays.add(delay);

				// 获取这段时间内并发用户数的最大值
				int tempThread = jsonObject.getIntValue("threadCount");
				if (threadCount < tempThread) {
					threadCount = tempThread;
				}

				// 计算这段时间内总的发送MB
				sendBytes = sendBytes + jsonObject.getDoubleValue("sentBytes");

				// 计算这段时间内总的接受MB
				receiveBytes = receiveBytes + jsonObject.getDoubleValue("receiveBytes");
			}

			if (successCount + failCount <= 0) {
				System.out.println("");
				continue;
			}

			Collections.sort(delays);

			avgDelay = totalDelay / (successCount + failCount);
			qps = (successCount + failCount) / (interval / 1000);

			resObj.put("timeStamp", endStamp);
			resObj.put("label", key);
			resObj.put("successCount", successCount);
			resObj.put("failCount", failCount);
			resObj.put("avgTime", avgDelay);
			resObj.put("qps", qps);
			resObj.put("90%Delay", delays.get(length_level1));
			resObj.put("95%Delay", delays.get(length_level2));
			resObj.put("99%Delay", delays.get(length_level3));
			resObj.put("minDelay", delays.get(0));
			resObj.put("maxDelay", delays.get(length - 1));
			resObj.put("sendBytes/s", sendBytes / (interval / 1000));
			resObj.put("receiveBytes/s", receiveBytes / (interval / 1000));
			resObj.put("thread", threadCount);

			resArry.add(resObj);

			// 添加到TOTAL计数中
			TOTAL_successCount = TOTAL_successCount + successCount;
			TOTAL_failCount = TOTAL_failCount + failCount;
			TOTAL_totalDelay = TOTAL_totalDelay + totalDelay;
			TOTAL_delays.addAll(delays);
			TOTAL_sendBytes = TOTAL_sendBytes + sendBytes;
			TOTAL_receiveBytes = TOTAL_receiveBytes + receiveBytes;
			if (TOTAL_threadCount < threadCount) {
				TOTAL_threadCount = threadCount;
			}

			TOTAL_length = TOTAL_length + length;
		}

		TOTAL_length_level1 = TOTAL_length * 90 / 100 - 1;
		TOTAL_length_level2 = TOTAL_length * 95 / 100 - 1;
		TOTAL_length_level3 = TOTAL_length * 99 / 100 - 1;

		if (TOTAL_length_level1 > 0) {
			Collections.sort(TOTAL_delays);

			TOTAL_avgDelay = TOTAL_totalDelay / (TOTAL_successCount + TOTAL_failCount);
			TOTAL_qps = (TOTAL_successCount + TOTAL_failCount) / (interval / 1000);

			total.put("timeStamp", endStamp);
			total.put("label", "Total");
			total.put("successCount", TOTAL_successCount);
			total.put("failCount", TOTAL_failCount);
			total.put("avgTime", TOTAL_avgDelay);
			total.put("qps", TOTAL_qps);
			total.put("90%Delay", TOTAL_delays.get(TOTAL_length_level1));
			total.put("95%Delay", TOTAL_delays.get(TOTAL_length_level2));
			total.put("99%Delay", TOTAL_delays.get(TOTAL_length_level3));
			total.put("minDelay", TOTAL_delays.get(0));
			total.put("maxDelay", TOTAL_delays.get(TOTAL_length - 1));
			total.put("sendBytes/s", TOTAL_sendBytes / (interval / 1000));
			total.put("receiveBytes/s", TOTAL_receiveBytes / (interval / 1000));
			total.put("thread", TOTAL_threadCount);

			resArry.add(total);
		}

		System.gc();

		return resArry;
	}
}
